package DianShang_2024.ds_03.tzgc

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.dense_rank

import java.util.Properties

object tzgc01 {
  def main(args: Array[String]): Unit = {
    /*
        1、据Hive的dwd库中相关表或MySQL数据库shtd_store中订单相关表（order_detail、order_info、sku_info），对用户购
        买过的商品进行去重，将其转换为以下格式：第一列为用户id mapping，第二列为用户购买过的商品id mapping，按
        照user_id与sku_id进行升序排序，输出前5行，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的
        任务序号下；
    字段	类型	中文含义	备注
    user_id	int	用户id的mapping对应键
    sku_id	int	商品id的mapping对应键
    提示：
    Mapping操作：例如用户id：1、4、7、8、9，则做完mapping操作转为字典类型，键0对应用户id 1，键1对应用户id 4，以此类推
    结果格式如下：
    -------user_id_mapping与sku_id_mapping数据前5条如下：-------
    0:0
    0:89
    1:1
    1:2
    1:3
     */
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("第三套卷子的特征工程第一题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()


    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")


    val order_info=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_info",connect)
    val order_detail=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_detail",connect)



    //  使用dense_rank的效果和StringIndexer的效果是一样的，而且使用rank是不行的，因为dense_rank如果值一样的话排名也就一样，并且不会占据后面的位
    //  置，也就是112334，但是使用rank就是1134
    val result=order_info.join(order_detail,order_detail("order_id")===order_info("id"))
      .select("user_id","sku_id")
      .withColumn(
        "user_mapping",
        dense_rank().over(Window.orderBy("user_id")) -1
      )
      .withColumn(
        "sku_mapping",
        dense_rank().over(Window.orderBy("sku_id")) -1
      )
      .orderBy("user_id","sku_id")
      .limit(5)
      .select("user_mapping","sku_mapping")
      .collect()


      println("--------------------user_id_mapping和sku_id_mapping数据前5条如下:----------------")
      result.foreach{
        (r) => println(r.toSeq.mkString(":"))
      }



    spark.close()

  }

}
